package io.debezium.connector.postgresql;

import com.huawei.gaussdb.jdbc.core.BaseConnection;
import io.debezium.connector.postgresql.connection.GaussDBConnection;
import io.debezium.connector.postgresql.connection.ReplicationMessage;
import io.debezium.data.Envelope;
import io.debezium.data.Envelope.Operation;
import io.debezium.function.Predicates;
import io.debezium.pipeline.spi.ChangeRecordEmitter;
import io.debezium.pipeline.spi.OffsetContext;
import io.debezium.relational.*;
import io.debezium.schema.DataCollectionSchema;
import io.debezium.util.Clock;
import io.debezium.util.Strings;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.header.ConnectHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.SQLException;
import java.util.*;
import java.util.stream.Collectors;

public class PostgresChangeRecordEmitter extends RelationalChangeRecordEmitter<PostgresPartition> {
    private static final Logger LOGGER = LoggerFactory.getLogger(PostgresChangeRecordEmitter.class);
    private final ReplicationMessage message;
    private final PostgresSchema schema;
    private final PostgresConnectorConfig connectorConfig;
    private final GaussDBConnection connection;
    private final TableId tableId;
    private final boolean unchangedToastColumnMarkerMissing;
    private final boolean nullToastedValuesMissingFromOld;
    private final Map<String, Object> cachedOldToastedValues = new HashMap();

    public PostgresChangeRecordEmitter(
            PostgresPartition partition,
            OffsetContext offset,
            Clock clock,
            PostgresConnectorConfig connectorConfig,
            PostgresSchema schema,
            GaussDBConnection connection,
            TableId tableId,
            ReplicationMessage message) {
        super(partition, offset, clock);
        this.schema = schema;
        this.message = message;
        this.connectorConfig = connectorConfig;
        this.connection = connection;
        this.tableId = tableId;
        this.unchangedToastColumnMarkerMissing =
                !connectorConfig.plugin().hasUnchangedToastColumnMarker();
        this.nullToastedValuesMissingFromOld =
                !connectorConfig.plugin().sendsNullToastedValuesInOld();
        Objects.requireNonNull(this.tableId);
    }

    public Envelope.Operation getOperation() {
        switch (this.message.getOperation()) {
            case INSERT:
                return Operation.CREATE;
            case UPDATE:
                return Operation.UPDATE;
            case DELETE:
                return Operation.DELETE;
            case TRUNCATE:
                return Operation.TRUNCATE;
            default:
                throw new IllegalArgumentException(
                        "Received event of unexpected command type: "
                                + this.message.getOperation());
        }
    }

    public void emitChangeRecords(
            DataCollectionSchema schema, ChangeRecordEmitter.Receiver<PostgresPartition> receiver)
            throws InterruptedException {
        schema = this.synchronizeTableSchema(schema);
        super.emitChangeRecords(schema, receiver);
    }

    protected void emitTruncateRecord(
            ChangeRecordEmitter.Receiver receiver, TableSchema tableSchema)
            throws InterruptedException {
        Struct envelope =
                tableSchema
                        .getEnvelopeSchema()
                        .truncate(
                                this.getOffset().getSourceInfo(),
                                this.getClock().currentTimeAsInstant());
        receiver.changeRecord(
                this.getPartition(),
                tableSchema,
                Operation.TRUNCATE,
                (Object) null,
                envelope,
                this.getOffset(),
                (ConnectHeaders) null);
    }

    protected Object[] getOldColumnValues() {
        try {
            switch (this.getOperation()) {
                case CREATE:
                    return null;
                case UPDATE:
                    return this.columnValues(
                            this.message.getOldTupleList(),
                            this.tableId,
                            true,
                            this.message.hasTypeMetadata(),
                            true,
                            true);
                default:
                    return this.columnValues(
                            this.message.getOldTupleList(),
                            this.tableId,
                            true,
                            this.message.hasTypeMetadata(),
                            false,
                            true);
            }
        } catch (SQLException var2) {
            SQLException e = var2;
            throw new ConnectException(e);
        }
    }

    protected Object[] getNewColumnValues() {
        try {
            switch (this.getOperation()) {
                case CREATE:
                    return this.columnValues(
                            this.message.getNewTupleList(),
                            this.tableId,
                            true,
                            this.message.hasTypeMetadata(),
                            false,
                            false);
                case UPDATE:
                    return this.columnValues(
                            this.message.getNewTupleList(),
                            this.tableId,
                            true,
                            this.message.hasTypeMetadata(),
                            false,
                            false);
                default:
                    return null;
            }
        } catch (SQLException var2) {
            SQLException e = var2;
            throw new ConnectException(e);
        }
    }

    private DataCollectionSchema synchronizeTableSchema(DataCollectionSchema tableSchema) {
        if (this.getOperation() != Operation.DELETE && this.message.shouldSchemaBeSynchronized()) {
            boolean metadataInMessage = this.message.hasTypeMetadata();
            TableId tableId = (TableId) tableSchema.id();
            Table table = this.schema.tableFor(tableId);
            List<ReplicationMessage.Column> columns = this.message.getNewTupleList();
            if (this.schemaChanged(columns, table, metadataInMessage)) {
                this.refreshTableFromDatabase(tableId);
                if (metadataInMessage) {
                    this.schema.refresh(
                            this.tableFromFromMessage(columns, this.schema.tableFor(tableId)));
                }
            }

            return this.schema.schemaFor(tableId);
        } else {
            return tableSchema;
        }
    }

    private Object[] columnValues(
            List<ReplicationMessage.Column> columns,
            TableId tableId,
            boolean refreshSchemaIfChanged,
            boolean metadataInMessage,
            boolean sourceOfToasted,
            boolean oldValues)
            throws SQLException {
        if (columns != null && !columns.isEmpty()) {
            Table table = this.schema.tableFor(tableId);
            Objects.requireNonNull(table);
            List<Column> schemaColumns = table.columns();
            List<ReplicationMessage.Column> columnsWithoutToasted =
                    columns.stream()
                            .filter(Predicates.not(ReplicationMessage.Column::isToastedColumn))
                            .collect(Collectors.toList());
            Object[] values =
                    new Object[Math.max(columnsWithoutToasted.size(), schemaColumns.size())];
            Set<String> undeliveredToastableColumns =
                    new HashSet(this.schema.getToastableColumnsForTableId(table.id()));
            Iterator var12 = columns.iterator();

            while (var12.hasNext()) {
                ReplicationMessage.Column column = (ReplicationMessage.Column) var12.next();
                String columnName = Strings.unquoteIdentifierPart(column.getName());
                undeliveredToastableColumns.remove(columnName);
                int position = this.getPosition(columnName, table, values);
                if (position != -1) {
                    Object value =
                            column.getValue(
                                    () -> {
                                        return (BaseConnection) this.connection.connection();
                                    },
                                    this.connectorConfig.includeUnknownDatatypes());
                    if (sourceOfToasted) {
                        this.cachedOldToastedValues.put(columnName, value);
                    } else if (value
                            == UnchangedToastedReplicationMessageColumn.UNCHANGED_TOAST_VALUE) {
                        Object candidate = this.cachedOldToastedValues.get(columnName);
                        if (candidate != null) {
                            value = candidate;
                        }
                    }

                    values[position] = value;
                }
            }

            if (this.unchangedToastColumnMarkerMissing) {
                var12 = undeliveredToastableColumns.iterator();

                while (true) {
                    while (true) {
                        String columnName;
                        int position;
                        do {
                            if (!var12.hasNext()) {
                                return values;
                            }

                            columnName = (String) var12.next();
                            position = this.getPosition(columnName, table, values);
                        } while (position == -1);

                        Object candidate = this.cachedOldToastedValues.get(columnName);
                        if (oldValues && this.nullToastedValuesMissingFromOld) {
                            values[position] = null;
                        } else {
                            values[position] =
                                    candidate != null
                                            ? candidate
                                            : UnchangedToastedReplicationMessageColumn
                                                    .UNCHANGED_TOAST_VALUE;
                        }
                    }
                }
            } else {
                return values;
            }
        } else {
            return null;
        }
    }

    private int getPosition(String columnName, Table table, Object[] values) {
        Column tableColumn = table.columnWithName(columnName);
        if (tableColumn == null) {
            LOGGER.warn(
                    "Internal schema is out-of-sync with incoming decoder events; column {} will be omitted from the change event.",
                    columnName);
            return -1;
        } else {
            int position = tableColumn.position() - 1;
            if (position >= 0 && position < values.length) {
                return position;
            } else {
                LOGGER.warn(
                        "Internal schema is out-of-sync with incoming decoder events; column {} will be omitted from the change event.",
                        columnName);
                return -1;
            }
        }
    }

    private Optional<DataCollectionSchema> newTable(TableId tableId) {
        LOGGER.debug("Schema for table '{}' is missing", tableId);
        this.refreshTableFromDatabase(tableId);
        TableSchema tableSchema = this.schema.schemaFor(tableId);
        if (tableSchema == null) {
            LOGGER.warn("cannot load schema for table '{}'", tableId);
            return Optional.empty();
        } else {
            LOGGER.debug("refreshed DB schema to include table '{}'", tableId);
            return Optional.of(tableSchema);
        }
    }

    private void refreshTableFromDatabase(TableId tableId) {
        try {
            this.schema.refresh(
                    this.connection,
                    tableId,
                    this.connectorConfig.skipRefreshSchemaOnMissingToastableData());
        } catch (SQLException var3) {
            SQLException e = var3;
            throw new ConnectException("Database error while refresing table schema", e);
        }
    }

    static Optional<DataCollectionSchema> updateSchema(
            PostgresPartition partition,
            TableId tableId,
            ChangeRecordEmitter<PostgresPartition> changeRecordEmitter) {
        return ((PostgresChangeRecordEmitter) changeRecordEmitter).newTable(tableId);
    }

    private boolean schemaChanged(
            List<ReplicationMessage.Column> columns, Table table, boolean metadataInMessage) {
        int tableColumnCount = table.columns().size();
        int replicationColumnCount = columns.size();
        boolean msgHasMissingColumns = tableColumnCount > replicationColumnCount;
        if (msgHasMissingColumns
                && this.connectorConfig.skipRefreshSchemaOnMissingToastableData()) {
            msgHasMissingColumns = this.hasMissingUntoastedColumns(table, columns);
        }

        boolean msgHasAdditionalColumns = tableColumnCount < replicationColumnCount;
        if (!msgHasMissingColumns && !msgHasAdditionalColumns) {
            return columns.stream()
                    .anyMatch(
                            (message) -> {
                                String columnName = message.getName();
                                Column column = table.columnWithName(columnName);
                                if (column == null) {
                                    LOGGER.info(
                                            "found new column '{}' present in the server message which is not part of the table metadata; refreshing table schema",
                                            columnName);
                                    return true;
                                } else {
                                    int localType = column.nativeType();
                                    int incomingType = message.getType().getOid();
                                    int localLength;
                                    if (localType != incomingType) {
                                        localLength = message.getType().getRootType().getOid();
                                        if (localType != localLength) {
                                            LOGGER.info(
                                                    "detected new type for column '{}', old type was {} ({}), new type is {} ({}); refreshing table schema",
                                                    new Object[] {
                                                        columnName,
                                                        localType,
                                                        column.typeName(),
                                                        incomingType,
                                                        message.getType().getName()
                                                    });
                                            return true;
                                        }
                                    }

                                    if (metadataInMessage) {
                                        localLength = column.length();
                                        int incomingLength = message.getTypeMetadata().getLength();
                                        if (localLength != incomingLength) {
                                            LOGGER.info(
                                                    "detected new length for column '{}', old length was {}, new length is {}; refreshing table schema",
                                                    new Object[] {
                                                        columnName, localLength, incomingLength
                                                    });
                                            return true;
                                        }

                                        int localScale =
                                                (Integer)
                                                        column.scale()
                                                                .orElseGet(
                                                                        () -> {
                                                                            return 0;
                                                                        });
                                        int incomingScale = message.getTypeMetadata().getScale();
                                        if (localScale != incomingScale) {
                                            LOGGER.info(
                                                    "detected new scale for column '{}', old scale was {}, new scale is {}; refreshing table schema",
                                                    new Object[] {
                                                        columnName, localScale, incomingScale
                                                    });
                                            return true;
                                        }

                                        boolean localOptional = column.isOptional();
                                        boolean incomingOptional = message.isOptional();
                                        if (localOptional != incomingOptional) {
                                            LOGGER.info(
                                                    "detected new optional status for column '{}', old value was {}, new value is {}; refreshing table schema",
                                                    new Object[] {
                                                        columnName, localOptional, incomingOptional
                                                    });
                                            return true;
                                        }
                                    }

                                    return false;
                                }
                            });
        } else {
            LOGGER.info(
                    "Different column count {} present in the server message as schema in memory contains {}; refreshing table schema",
                    replicationColumnCount,
                    tableColumnCount);
            return true;
        }
    }

    private boolean hasMissingUntoastedColumns(
            Table table, List<ReplicationMessage.Column> columns) {
        List<String> msgColumnNames =
                columns.stream()
                        .map(ReplicationMessage.Column::getName)
                        .collect(Collectors.toList());
        List<String> missingColumnNames =
                table.columns().stream()
                        .filter(
                                (c) -> {
                                    return !msgColumnNames.contains(c.name());
                                })
                        .map(Column::name)
                        .collect(Collectors.toList());
        List<String> toastableColumns = this.schema.getToastableColumnsForTableId(table.id());
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(
                    "msg columns: '{}' --- missing columns: '{}' --- toastableColumns: '{}",
                    new Object[] {
                        String.join(",", msgColumnNames),
                        String.join(",", missingColumnNames),
                        String.join(",", toastableColumns)
                    });
        }

        return !toastableColumns.containsAll(missingColumnNames);
    }

    private Table tableFromFromMessage(List<ReplicationMessage.Column> columns, Table table) {
        TableEditor combinedTable =
                table.edit()
                        .setColumns(
                                columns.stream()
                                        .map(
                                                (column) -> {
                                                    PostgresType type = column.getType();
                                                    ColumnEditor columnEditor =
                                                            Column.editor()
                                                                    .name(column.getName())
                                                                    .jdbcType(
                                                                            type.getRootType()
                                                                                    .getJdbcId())
                                                                    .type(type.getName())
                                                                    .optional(column.isOptional())
                                                                    .nativeType(
                                                                            type.getRootType()
                                                                                    .getOid());
                                                    columnEditor.length(
                                                            column.getTypeMetadata().getLength());
                                                    columnEditor.scale(
                                                            column.getTypeMetadata().getScale());
                                                    Optional<String> var10000 =
                                                            Optional.ofNullable(
                                                                            table.columnWithName(
                                                                                    column
                                                                                            .getName()))
                                                                    .flatMap(
                                                                            Column
                                                                                    ::defaultValueExpression);
                                                    Objects.requireNonNull(columnEditor);
                                                    var10000.ifPresent(
                                                            columnEditor::defaultValueExpression);
                                                    return columnEditor.create();
                                                })
                                        .collect(Collectors.toList()));
        List<String> pkCandidates = new ArrayList(table.primaryKeyColumnNames());
        Iterator<String> itPkCandidates = pkCandidates.iterator();

        while (itPkCandidates.hasNext()) {
            String candidateName = (String) itPkCandidates.next();
            if (!combinedTable.hasUniqueValues()
                    && combinedTable.columnWithName(candidateName) == null) {
                LOGGER.error("Potentional inconsistency in key for message {}", columns);
                itPkCandidates.remove();
            }
        }

        combinedTable.setPrimaryKeyNames(pkCandidates);
        return combinedTable.create();
    }

    protected boolean skipEmptyMessages() {
        return true;
    }
}
